package com.atguigu.flink.chapter11.function;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2022/1/23 11:06
 */
public class Flink02_Function_Table {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        DataStreamSource<String> stream = env.fromElements("hello world", "hello hello", "hello atguigu world");
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        Table table = tEnv.fromDataStream(stream, $("line"));
        tEnv.createTemporaryView("sensor", table);
        
        // 1. 在tableAPI中
        // 1.1 内联的方式使用
        /*table
            //            .joinLateral(call(MySplit.class, $("line")))   // 炸出来之后, 每列的列名是在定义的时候指定好的. 内连接
            .leftOuterJoinLateral(call(MySplit.class, $("line")).as("w", "L"))   // 炸出来之后, 每列的列名是在定义的时候指定好的. 内连接
            .select($("line"), $("w"), $("L"))
            .execute()
            .print();
        */
        // 1.2 先注册后使用
        
        
        // 2. 在sql语句中使用
        tEnv.createTemporaryFunction("my_split", MySplit.class);
        // select   from a join b on a.id=b.id
        // select   from a,b where a.id=b.id   都是内连接
        /*tEnv.sqlQuery("select" +
                          " line," +
                          " word," +
                          " len " +
                          "from sensor " +
                          "join lateral table(my_split(line)) on true").execute().print(); // 连接条件只能数true
        */
        tEnv.sqlQuery("select" +
                          " line," +
                          " word," +
                          " len " +
                          "from sensor " +
                          ", lateral table(my_split(line))").execute().print(); // 连接条件只能数true*/
    
       /* tEnv.sqlQuery("select" +
                          " line," +
                          " w," +
                          " L " +
                          "from sensor " +
                          "left join lateral table(my_split(line)) as t(w, L) on true").execute().print();*/ // 连接条件只能数true*/
    
    
    
    }
    
    @FunctionHint(output = @DataTypeHint("row<word string, len int>"))
    public static class MySplit extends TableFunction<Row> {
        public void eval(String line) {
            if (line.contains("atguigu")) {
                return;
            }
            
            String[] words = line.split(" ");
            // 每调用一次, 就相当于产生一行数据
            for (String word : words) {
                collect(Row.of(word, word.length()));
            }
            
        }
    }
    
    /*public static class MySplit extends TableFunction<WordLen> {
        public void eval(String line) {
            if (line.contains("atguigu")) {
                return;
            }
            
            String[] words = line.split(" ");
            // 每调用一次, 就相当于产生一行数据
            for (String word : words) {
                collect(new WordLen(word, (long) word.length()));
            }
            
        }
    }
    
    public static class WordLen {
        
        public WordLen(String word, Long len) {
            this.word = word;
            this.len = len;
        }
        
        public String word;
        public Long len;
        
    }*/
    
}
/*
"hello world"
                hello     5
                world     5

 */